package com.leiyuee.flink.tools.flink;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.concurrent.TimeUnit;

public class StreamEnv {

    /**
     * 纯净的环境，Parallelism = 1
     * @return StreamExecutionEnvironment
     */
    public static StreamExecutionEnvironment getRealtimeEnv() {
        return getRealtimeEnv(false, null);
    }


    /**
     * 仅带配置的环境
     * @param hasConfig 是否设置配置
     * @return StreamExecutionEnvironment
     */
    public static StreamExecutionEnvironment getRealtimeEnv(boolean hasConfig) {
        return getRealtimeEnv(hasConfig, null);
    }

    /**
     * 仅带检查点的环境
     * @param checkpointDirectory 检查点路径
     * @return StreamExecutionEnvironment
     */
    public static StreamExecutionEnvironment getRealtimeEnv(String checkpointDirectory) {
        return getRealtimeEnv(false, checkpointDirectory);
    }

    /**
     * 获取环境
     * @param hasConfig 是否设置环境配置
     * @param checkpointDirectory 配置检查点目录，否则去除检查点
     * @return StreamExecutionEnvironment
     */
    public static StreamExecutionEnvironment getRealtimeEnv(boolean hasConfig, String checkpointDirectory) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 获取带配置的环境
        if (hasConfig) {
            env = getRealtimeEnvHasConfig();
        }

        // 给环境设置检查点
        if (checkpointDirectory != null && !checkpointDirectory.isEmpty()) {
            setEnvCheckpoint(env, checkpointDirectory);
        }

        env.setParallelism(1);
        return env;
    }


    /**
     * 获取环境
     * @return StreamExecutionEnvironment
     */
    private static StreamExecutionEnvironment getRealtimeEnvHasConfig() {
        Configuration envConf = new Configuration();
        envConf.setString("parallelism.default", "1");
        envConf.setString("table.local-time-zone", "Asia/Shanghai");

        envConf.setString("execution.checkpointing.interval", "60000");
        envConf.setString("execution.checkpointing.externalized-checkpoint-retention", "RETAIN_ON_CANCELLATION");
        envConf.setString("execution.checkpointing.max-concurrent-checkpoints", "1");
        envConf.setString("execution.checkpointing.min-pause", "1000");
        envConf.setString("execution.checkpointing.mode", "EXACTLY_ONCE");
        envConf.setString("execution.checkpointing.timeout", "300000");
        envConf.setString("execution.checkpointing.tolerable-failed-checkpoints", "10");

        envConf.setString("state.checkpoints.num-retained", "3");
        envConf.setString("state.backend.incremental", "true");

        envConf.setString("restart-strategy", "fixed-delay");
        envConf.setString("restart-strategy.fixed-delay.attempts", "3");
        envConf.setString("restart-strategy.fixed-delay.delay", "10 s");

        envConf.setString("execution.runtime-mode", "streaming");
        envConf.setString("sql-client.execution.result-mode", "tableau");

        envConf.setString("pipeline.time-characteristic", "EventTime");
        envConf.setString("pipeline.auto-watermark-interval", "1000");
        envConf.setString("execution.buffer-timeout.enabled", "true");
        envConf.setString("execution.buffer-timeout.interval", "2000");
        return StreamExecutionEnvironment.getExecutionEnvironment(envConf);
    }

    /**
     * 给环境设置检查点
     * @param env 需要设置检查点的环境
     * @param checkpointDirectory 检查点目录路径
     */
    private static void setEnvCheckpoint(StreamExecutionEnvironment env, String checkpointDirectory) {
        env.setRestartStrategy(RestartStrategies.failureRateRestart(10, Time.of(10L, TimeUnit.MINUTES), Time.of(10L, TimeUnit.SECONDS)));
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(60000L);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage(checkpointDirectory));
        System.setProperty("HADOOP_USER_NAME", "jabil");
        env.enableCheckpointing(60000L);
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(5);// 故障恢复
    }

}
